package com.manage.test.web.demo;

import cn.hutool.json.JSONUtil;
import com.manage.test.web.dto.MessageSendDto;
import lombok.RequiredArgsConstructor;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.codec.SerializationCodec;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.function.Consumer;

/**
 * @Author: zhangqing
 * @CreateTime: 2023-03-09 21:31
 * @Description: 在这里添加描述
 */
@RestController
@RequestMapping("/server")
@RequiredArgsConstructor
public class Server {

    private final RedissonClient redissonClient;

    private static Integer num = 10;

    public static void main(String[] args) throws Exception {
        // 没有返回值的异步任务
//        runAsync();
        // 有返回值的异步任务
//        String s = supplyAsync();
        // 线程依赖 [当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。]
//        Integer integer = thenApply();
        // 消费处理结果 [消费处理结果, 接收任务的处理结果，并消费处理，无返回结果。]
//        thenAccept();
        // 合并两个没有依赖关系的 CompletableFutures 任务
        thenCombine();
    }

    public static void thenCombine() throws ExecutionException, InterruptedException {
        System.out.println("主线程开始");
        CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("加 10 任务开始");
            num += 10;
            return num;
        });
        CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("乘以 10 任务开始");
            num = num * 10;
            return num;
        });
        //合并两个结果
        CompletableFuture<Object> future = job1.thenCombine(job2, (BiFunction<Integer, Integer, List<Integer>>) (a, b) -> {
            List<Integer> list = new ArrayList<>();
            list.add(a);
            list.add(b);
            return list;
        });
        System.out.println("合并结果为:" + future.get());
    }

    public static void thenAccept(){
        AtomicInteger num = new AtomicInteger(10);
        System.out.println("主线程开始");

        long start = System.currentTimeMillis();
        System.out.println("开始时间" + start);

        CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("加 10 任务开始");
                num.addAndGet(10);
                Thread.sleep(5000);
                long futureEnd = System.currentTimeMillis();
                System.out.println("加 10 任务完成耗时" + (futureEnd - start));
            } catch (Exception e) {
                e.printStackTrace();
            }
            return num;
        }).thenApply(integer -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            long futureEnd1 = System.currentTimeMillis();
            System.out.println("取平方任务完成耗时" + (futureEnd1 - start));
            return num.get() * num.get();
        }).thenAccept(integer -> System.out.println("执行全部结束耗时：" + (System.currentTimeMillis() - start)));
    }

    public static Integer thenApply() throws ExecutionException, InterruptedException {
        AtomicInteger num = new AtomicInteger(10);
        System.out.println("主线程开始");
        long start = System.currentTimeMillis();
        System.out.println("开始时间" + start);

        CompletableFuture<Integer> future =
                CompletableFuture.supplyAsync(() -> {
                    try {
                        System.out.println("加 10 任务开始");
                        num.addAndGet(10);
                        Thread.sleep(5000);
                        long futureEnd = System.currentTimeMillis();
                        System.out.println("加 10 任务完成耗时" + (futureEnd - start));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    return num;
                }).thenApply(integer -> {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    long futureEnd1 = System.currentTimeMillis();
                    System.out.println("取平方任务完成耗时" + (futureEnd1 - start));
                    return num.get() * num.get();
                });
        Integer integer = future.get();
        System.out.println("主线程结束, 子线程的结果为:" + integer);
        System.out.println("执行全部结束耗时：" + (System.currentTimeMillis() - start));
        return integer;
    }

    /**
     * 有返回值的异步任务
     */
    public static String supplyAsync() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        System.out.println("开始时间" + start);

        System.out.println("主线程开始");
        //运行一个有返回值的异步任务
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("子线程1开始任务");
                Thread.sleep(5000);
                long futureEnd = System.currentTimeMillis();
                System.out.println("子线程1完成耗时" + (futureEnd - start));
            } catch (Exception e) {
                e.printStackTrace();
            }
            return "子线程1完成了!";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("子线程2开始任务");
                Thread.sleep(5000);
                long futureEnd1 = System.currentTimeMillis();
                System.out.println("子线程2完成耗时" + (futureEnd1 - start));
            } catch (Exception e) {
                e.printStackTrace();
            }
            return "子线程2完成了!";
        });
        //主线程阻塞
        String s1 = future1.get();
        String s2 = future2.get();
        System.out.println("主线程结束, 子线程的结果为:" + s1+s2);
        System.out.println("执行全部结束耗时：" + (System.currentTimeMillis() - start));
        return s1+s2;
    }

    public static void runAsync() throws ExecutionException, InterruptedException {

        long start = System.currentTimeMillis();
        System.out.println("开始时间" + start);

        System.out.println("主线程开始");
        //运行一个没有返回值的异步任务
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            try {
                System.out.println("future->子线程启动干活");
                Thread.sleep(5000);
                long futureEnd = System.currentTimeMillis();
                System.out.println("future->子线程完成耗时" + (futureEnd - start));
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            try {
                System.out.println("future1->子线程启动干活");
                Thread.sleep(5000);
                long futureEnd1 = System.currentTimeMillis();
                System.out.println("future1->子线程完成耗时" + (futureEnd1 - start));
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        //主线程阻塞
        future.get();
        future1.get();
        System.out.println("主线程结束");
        System.out.println("执行全部结束耗时：" + (System.currentTimeMillis() - start));
    }

    @GetMapping("/send")
    public void send() {
        RTopic test = redissonClient.getTopic("test", new SerializationCodec());
        MessageSendDto m = MessageSendDto.builder().message("我是消息").toUserId(10L).type("1").build();
        String s = JSONUtil.toJsonStr(m);
        test.publish(s);
        System.out.println("发送消息成功");
    }

}
